/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.nio.ByteBuffer;

import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeRangeSearchCursor;
import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexCursor;
import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;
import edu.ucsd.db.bassarisk.storage.common.HyracksBtreeLexicon;

public class HyracksBTreeLexiconReader extends HyracksBtreeLexicon implements
    ILexiconReader {

  private ITreeIndexAccessor btreeAccessor;
  private ITreeIndexCursor btreeCursor;
  private RangePredicate btreePred;
  private final ByteBuffer frame;
  private FrameTupleReference searchKey;
  private FrameTupleAccessor tokenAccessor;
  private FrameTupleAppender tokenAppender;
  private ArrayTupleBuilder tokenTupleBuilder;

  public HyracksBTreeLexiconReader() {
    this.frame = ByteBuffer.allocate(ColumnStore.BYTES_IN_DRAWER);
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * edu.ucsd.db.bassarisk.storage.reading.ILexiconReader#findColumn(java.lang
   * .String)
   */
  @Override
  public int findColumn(final String token) {
    try {
      this.tokenTupleBuilder.reset();
      this.tokenTupleBuilder.addField(UTF8StringSerializerDeserializer.INSTANCE,
                                      token);

      this.tokenAppender.reset(this.frame, true);
      this.tokenAppender.append(this.tokenTupleBuilder.getFieldEndOffsets(),
                                this.tokenTupleBuilder.getByteArray(), 0,
                                this.tokenTupleBuilder.getSize());

      this.searchKey.reset(this.tokenAccessor, 0);

      this.btreeAccessor.search(this.btreeCursor, this.btreePred);

      if (this.btreeCursor.hasNext()) {

        this.btreeCursor.next();
        ITupleReference frameTuple = this.btreeCursor.getTuple();

        // hardcoded mapping of btree field
        int columnHeaderAddress =
            IntegerSerializerDeserializer.getInt(frameTuple.getFieldData(1),
                                                 frameTuple.getFieldStart(1));

        return columnHeaderAddress;
      } else {
        return -1;
      }
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  public void initialize(final IBufferCache bufferCache, final String lexFileName,
                         final IFileMapProvider fileMap) {
    try {
      super.initialize(bufferCache, lexFileName, fileMap, false);
    } catch (HyracksDataException e) {
      throw new RuntimeException(e);
    }

    // Set up various things for reading posting lists
    IBTreeLeafFrame leafFrame =
        (IBTreeLeafFrame) this.lexicon.getLeafFrameFactory().createFrame();
    this.btreeCursor = new BTreeRangeSearchCursor(leafFrame, false);
    this.searchKey = new FrameTupleReference();
    this.btreePred =
        new RangePredicate(true, this.searchKey, this.searchKey, true, true,
                           this.lexicon.getMultiComparator(),
                           this.lexicon.getMultiComparator());

    RecordDescriptor tokenRecDesc =
        new RecordDescriptor(
                             new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE });
    this.tokenAppender = new FrameTupleAppender(ColumnStore.BYTES_IN_DRAWER);

    this.tokenTupleBuilder = new ArrayTupleBuilder(1);
    this.tokenAccessor =
        new FrameTupleAccessor(ColumnStore.BYTES_IN_DRAWER, tokenRecDesc);
    this.tokenAccessor.reset(this.frame);

    this.btreeAccessor = this.lexicon.createAccessor();
  }

}
